package com.blink.curator.recipe.distributed;

import com.blink.base.BaseTestSample;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.barriers.DistributedBarrier;
import org.apache.curator.framework.recipes.barriers.DistributedDoubleBarrier;
import org.junit.jupiter.api.Test;

import java.util.concurrent.*;

/**
 * <Description> 使用Curator实现分布式Barrier</Description>
 * <ClassName> DistributedBarrierSample</ClassName>
 *
 * @Author liuxianzhao
 * @Date 2018年01月13日 18:47
 */
public class DistributedBarrierSample extends BaseTestSample {
    private static CyclicBarrier barrier = new CyclicBarrier(3);
    static String barrier_path = "/curator_recipes_barrier_path";
    private DistributedBarrier distributedBarrier;

    @Test
    public void testCyclicBarrier() {
        ExecutorService executor = new ThreadPoolExecutor(3, 3,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
        executor.submit(new Runner("1号选手"));
        executor.submit(new Runner("2号选手"));
        executor.submit(new Runner("3号选手"));
        executor.shutdown();
        sleep(10);
    }

    @Test
    public void testDistributedBarrier() throws Exception {
        ExecutorService executor = new ThreadPoolExecutor(4, 8,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>());
        for (int i = 0; i < 5; i++) {
            new Thread(new DistributedRunner((i + 1) + "号选手")).start();
        }
        while (true) {
//            logger.info("0");
            if (distributedBarrier != null) {
                logger.info("1");
                distributedBarrier.removeBarrier();
                break;
            }
        }
//        sleep();
        executor.shutdown();
    }


    @Test
    public void testDistributedDoubleBarrier() {
        String barrier_path = "/curator_recipes_barrier_path";
        for (int i = 0; i < 5; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        CuratorFramework client = getZkCurator();
                        client.start();
                        DistributedDoubleBarrier barrier = new DistributedDoubleBarrier(client, barrier_path, 5);
                        Thread.sleep(Math.round(Math.random() * 3000));
                        logger.info(Thread.currentThread().getName() + "号进入barrier");
                        barrier.enter();
                        logger.info("启动...");
                        Thread.sleep(Math.round(Math.random() * 3000));
                        barrier.leave();
                        logger.info("退出...");
                    } catch (Exception e) {
                    }
                }
            }).start();
        }
        sleep();
    }


    class Runner implements Runnable {
        private String name;

        public Runner(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            logger.info(name + " 准备好了.");
            try {
                barrier.await();
            } catch (Exception e) {
            }
            logger.info(name + " 起跑!");
        }
    }

    class DistributedRunner implements Runnable {
        private String name;

        public DistributedRunner(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            CuratorFramework client = getZkCurator();
            client.start();
            distributedBarrier = new DistributedBarrier(client, barrier_path);
            try {
                distributedBarrier.setBarrier();
                logger.info(name + "准备好了!");
                distributedBarrier.waitOnBarrier();
                logger.info(name + "起跑!");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
